package org.wj.config.avg.learn.kafka;

import org.apache.kafka.clients.producer.*;

import java.util.Properties;
import java.util.concurrent.ExecutionException;
import java.util.concurrent.Future;

public class MyKafkaProducer {

    public void sendMessage() throws ExecutionException, InterruptedException {
        Properties properties = new Properties();
        properties.put("bootstrap.servers", "192.168.124.157:9092");
        properties.put("key.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("value.serializer", "org.apache.kafka.common.serialization.StringSerializer");
        properties.put("acks", "all");
        properties.put("retries", "3");
        properties.put("batch.size", 323840);
        properties.put("linger.ms", 10);
        properties.put("buffer.memory", 33554432);
        properties.put("max.block.ms", 3000);
        properties.put(ProducerConfig.PARTITIONER_CLASS_CONFIG, "org.wj.learn.kafka.AuditPartitioner");
        Producer<String, String> producer = new KafkaProducer<>(properties);
        for (int i = 0; i < 100; i++) {
            Future<RecordMetadata> send = producer.send(new ProducerRecord<>("my-topic", Integer.toString(i), Integer.toString(i)));
            try {
                System.out.println(send.get().toString());
            } catch (InterruptedException | ExecutionException e) {
                e.printStackTrace();
            }
        }
        producer.close();

    }

    public static void main(String[] args) throws ExecutionException, InterruptedException {
        MyKafkaProducer myKafkaProducer = new MyKafkaProducer();
        myKafkaProducer.sendMessage();
    }
}
